﻿using Confluent.Kafka;
 
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using MySql.Data.MySqlClient;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Ccteg.DB
{

    public class KafDataserviceModel
    {
        public string DataType { get; set; }
        public string OperateType { get; set; }
        public Data Data { get; set; }
    }
    public class Data
    {
        public string TaskNO { get; set; }
        public string State { get; set; }


    }
    public class KafDataservice
    {
        /// <summary>
        /// 处理二级单位上报的任务
        /// </summary>
        /// <param name="serviceProvider"></param>
        public static void ReportingInitialize(IServiceProvider serviceProvider)
        {
            // 在某些情况下，您可能需要在方法中解析其他服务。在这种情况下，
            //请确保在使用后释放服务。确保这一点的最佳方法是创建scoped服务，例如：
            //    return;

            string topic = "001013U";
            var conf = new ConsumerConfig
            {
                GroupId = "test-consumer-group2",
                BootstrapServers = "47.100.173.18:9092",
                AutoOffsetReset = AutoOffsetReset.Latest,
                EnableAutoCommit = false  // 设置非自动偏移，业务逻辑完成后手动处理偏移，防止数据丢失
            };
            Task.Factory.StartNew(() =>
            {
                using (var serviceScope = serviceProvider.CreateScope())
                {

                    var _dbContext = serviceScope.ServiceProvider.GetService<MySqlDbContext>(); //获取注入容器并且获取注入的DbConext
                    using (var consumer = new ConsumerBuilder<Null, string>(conf).Build())
                    {
                        // 订阅topic

                        consumer.Subscribe(topic);
                        while (true)
                        {
                           
                            ConsumeResult<Null, string> consumeResult = null;
                            try
                            {
                                // 开始我的业务逻辑 TimeSpan.FromMilliseconds(5000)
                                consumeResult = consumer.Consume();
                                string res = consumeResult.Message.Value;
                                var k = Newtonsoft.Json.JsonConvert.DeserializeObject<KafDataserviceModel>(res);
                                if (!string.IsNullOrEmpty(k.Data.TaskNO) && k.OperateType.ToLower() == "update")
                                {
                                    var entity = _dbContext.tsk_jobtask.FirstOrDefault(x => x.JobTaskID == int.Parse(k.Data.TaskNO));
                                    if (entity != null)
                                    {
                                        entity.TaskSchedule = "已完成";
                                       _dbContext.SaveChanges();
                                        consumer.Commit(consumeResult); //偏移量+1
                                    }

                                }
                               
                            }
                            catch (Exception ex)
                            {
                                string exs = ex.Message;
                                if (consumeResult != null)
                                {
                                    //consumer.Commit(consumeResult); //从队列中移除
                                    consumeResult = null;
                                }
                            }

                            Thread.Sleep(10);//防止cpu空转

                        }
                    }
                }
            });

        }

    }
}
